package com.atguigu.realtime.app.dwd;

import com.atguigu.realtime.app.BaseSQLApp;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class Dwd_UserTestApp extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_UserTestApp().init(
                3003,
                2,
                "Dwd_UserTestApp"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tabEnv) {
        readOdsDb(tabEnv, "Dwd_UserTestApp");
        readBaseDic(tabEnv);
        //TODO 过滤试卷表
        Table testPaper = tabEnv.sqlQuery("select " +
                "data[`id`] id, " +
                "data[`course_id`] course_id " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='test_paper' " +
                "and `type`='insert' ");
        tabEnv.createTemporaryView("test_paper", testPaper);

        //TODO 过滤试卷问题表
        Table testPaperQuestion = tabEnv.sqlQuery("select " +
                "data[`id`] id, " +
                "data[`paper_id`] paper_id, " +
                "data[`question_id`] question_id, " +
                "data[`score`] score " +
                "from test_paper_question  " +
                "where `database`='edu' " +
                "and `table`='test_paper_question' " +
                "and `type`='insert' ");
        tabEnv.createTemporaryView("test_paper_question", testPaperQuestion);

        //TODO 过滤问题选项表
        Table testQuestionOption = tabEnv.sqlQuery("select " +
                "data[`id`] id, " +
                "data[`question_id`] question_id, " +
                "data[`is_correct`] is_correct " +
                "from test_question_option " +
                "where `database`='edu' " +
                "and `table`='test_question_option' " +
                "and `type`='insert' ");
        tabEnv.createTemporaryView("test_question_option", testQuestionOption);

        tabEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));

        Table result = tabEnv.sqlQuery("select " +
                "tp.id, " +
                "tpq.question_id, " +
                "tpq.score, " +
                "tqo.is_correct " +
                "from test_paper_question tpq " +
                "left join test_paper tp " +
                "on tpq.paper_id = tp.id " +
                "left join test_question_option tqo " +
                "on tpq.question_id = tqo.question_id ");

        tabEnv.sqlQuery("select * from "+ result);
    }
}
